/*
 * Copyright Hundsun Technologies Inc. All Rights Reserved.
 */

package service

import (
	"context"
	"encoding/hex"
	"errors"
	"fmt"
	"strconv"
	"time"

	"github.com/golang/protobuf/proto"
	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/common/types"
	"hundsun.com/hsl/hschain/common/util"
	"hundsun.com/hsl/hschain/common/util/pb"
	types2 "hundsun.com/hsl/hschain/contract/jvm/types"
	nativeTypes "hundsun.com/hsl/hschain/contract/native/token/types"
	"hundsun.com/hsl/hschain/p2p/common/p2perror"
	"hundsun.com/hsl/hschain/protos"
	protoscommon "hundsun.com/hsl/hschain/protos/common"
	pbconfig "hundsun.com/hsl/hschain/protos/config"
	protosledger "hundsun.com/hsl/hschain/protos/ledger"
	protosp2p "hundsun.com/hsl/hschain/protos/p2p"
	protosrpc "hundsun.com/hsl/hschain/protos/rpc"
	ptypes "hundsun.com/hsl/hschain/protos/types"
	"hundsun.com/hsl/hschain/rpc/common/constant"
	"hundsun.com/hsl/hschain/rpc/common/rpcerror"
	"hundsun.com/hsl/hschain/rpc/common/utils"
	"hundsun.com/hsl/hschain/store/mq"
)

const (
	//LogCall ...
	LogCall = "call for method:%s, param:%v"

	//Pure 合约pure方法
	Pure = "Pure"
	//ReceiveTransaction 发送交易
	ReceiveTransaction = "ReceiveTransaction"
	//GetPeerCount 获取p2p节点数
	GetPeerCount = "GetPeerCount"
	//GetBlockNumber 获取块高
	GetBlockNumber = "GetBlockNumber"
	//GetBlockExecNumber 获取执行区块高度
	GetBlockExecNumber = "GetBlockExecNumber"
	//GetBalance 获取账户余额
	GetBalance = "GetBalance"
	//GetTransactionByHash 根据hash获取交易
	GetTransactionByHash = "GetTransactionByHash"
	//GetBlockByNumber 根据blockNumber获取区块
	GetBlockByNumber = "GetBlockByNumber"
	//GetExecutedBlockByNumber 根据blockNumber获取已执行的块
	GetExecutedBlockByNumber = "GetExecutedBlockByNumber"
	//GetExecutedBlocks 根据区块范围获取已执行的块
	GetExecutedBlocks = "GetExecutedBlocks"
	//GetTransactionTotal 获取交易总数
	GetTransactionTotal = "GetTransactionTotal"
	//GetContractTotal 获取合约总数
	GetContractTotal = "GetContractTotal"
	//GetContractVersion 获取合约版本
	GetContractVersion = "GetContractVersion"
	//GetAddressTotal 获取账户总数
	GetAddressTotal = "GetAccountTotal"
	//GetAccountList 获取账户列表
	GetAccountList = "GetAccountList"
	//GetHistoryBlockExeResult 获取区块执行结果
	GetHistoryBlockExeResult = "GetHistoryBlockExeResult"
	//GetHistoryContractEvents 查询历史合约事件
	GetHistoryContractEvents = "GetHistoryContractEvents"
	//P2pPrivateMessage 发送p2p点对点消息
	P2pPrivateMessage = "P2pPrivateMessage"
	//GetP2pPrivateMessage 获取p2p点对点消息
	GetP2pPrivateMessage = "GetP2pPrivateMessage"
	//PullP2pPrivateMessage 获取p2p点对点消息并删除
	PullP2pPrivateMessage = "PullP2pPrivateMessage"
	//DeleteP2pPrivateMessage 删除p2p点对点消息
	DeleteP2pPrivateMessage = "DeleteP2pPrivateMessage"
	//GetP2pPeerID 查询p2p peerId
	GetP2pPeerID = "GetP2pPeerID"
	//GetP2pPeerInfos 查询peerinfo
	GetP2pPeerInfos = "GetP2pPeerInfos"
	//GetConsensusInfo 查询共识相关信息
	GetConsensusInfo = "GetConsensusInfo"
	//IsSynced 查询blockchain同步状态
	IsSynced = "IsSynced"
	//GetLatestExecutedBlock 查询最新执行后的区块
	GetLatestExecutedBlock = "GetLatestExecutedBlock"
	//PeekTxPool 查询txpool中交易状态
	PeekTxPool = "PeekTxPool"
	//TxPoolSize 查询txpool大小
	TxPoolSize = "TxPoolSize"
	//TxPoolHashes 查询txpool中交易hash
	TxPoolHashes = "TxPoolHashes"

	//SubBlocks 订阅区块
	SubBlocks = "SubBlocks"
	//SubHistoryBlocks 订阅历史区块
	SubHistoryBlocks = "SubHistoryBlocks"
	//SubContractEvents 订阅合约事件
	SubContractEvents = "SubContractEvents"
	//SubHistoryEvents 订阅历史事件
	SubHistoryEvents = "SubHistoryEvents"
	//SubTransactionResult 订阅交易执行结果
	SubTransactionResult = "SubTransactionResult"
	//RPCMethodUnsubscribe 取消订阅
	RPCMethodUnsubscribe = "Unsubscribe"
	// QueryContract = 通用的查询合约方法
	QueryContract = "QueryContract"
)

//RPCMethodImpl rpc方法实现
type RPCMethodImpl struct {
	module *RPCService
	//ledger      ledger.Ledger
	GenesisConf    *pbconfig.GenesisTransactionPayload
	addChan        chan *protosledger.Transaction
	requestTimeOut time.Duration
}

//NewRPCMethodInstance 新建rpc方法实现
func NewRPCMethodInstance(module *RPCService) *RPCMethodImpl {
	methodInstance := &RPCMethodImpl{
		module:         module,
		GenesisConf:    pb.GetGenesisTransactionPayload(),
		addChan:        make(chan *protosledger.Transaction),
		requestTimeOut: time.Second * constant.MqTimeOut,
	}
	once.Do(
		func() {
			RegisterHandler(Pure, methodInstance.Pure)
			RegisterHandler(ReceiveTransaction, methodInstance.SendSignedTransactionAsyn)
			RegisterHandler(GetPeerCount, methodInstance.GetPeerCount)
			RegisterHandler(GetBlockNumber, methodInstance.GetBlockNumber)
			RegisterHandler(GetBlockExecNumber, methodInstance.GetBlockExecNumber)
			RegisterHandler(GetBalance, methodInstance.GetBalance)
			RegisterHandler(GetTransactionByHash, methodInstance.GetTransactionByHash)
			RegisterHandler(GetBlockByNumber, methodInstance.GetBlockByNumber)
			RegisterHandler(GetExecutedBlockByNumber, methodInstance.GetExecutedBlockByNumber)
			RegisterHandler(GetExecutedBlocks, methodInstance.GetExecutedBlocks)
			RegisterHandler(GetHistoryBlockExeResult, methodInstance.GetBlockExecResults)
			RegisterHandler(GetHistoryContractEvents, methodInstance.GetHistoryContractEvents)
			RegisterHandler(GetP2pPeerID, methodInstance.GetP2pPeerID)
			RegisterHandler(GetP2pPeerInfos, methodInstance.GetPeerInfos)
			RegisterHandler(GetConsensusInfo, methodInstance.GetConsensusInfo)
			RegisterHandler(IsSynced, methodInstance.IsSynced)
			RegisterHandler(GetLatestExecutedBlock, methodInstance.GetLatestExecutedBlock)
			RegisterHandler(GetContractVersion, methodInstance.GetContractVersion)
			RegisterHandler(PeekTxPool, methodInstance.PeekTxPool)
			RegisterHandler(TxPoolSize, methodInstance.TxPoolSize)
			RegisterHandler(TxPoolHashes, methodInstance.TxPoolHashes)
			RegisterHandler(QueryContract, methodInstance.QueryContract)
			go methodInstance.SendTransactionMonitor(module.ctx)
		})
	return methodInstance
}

//GetSubMethodAndParam 获取订阅方法名和入参
func (r *RPCMethodImpl) GetSubMethodAndParam(methodName string) (interface{}, interface{}, error) {
	switch methodName {
	case SubBlocks:
		return r.subBlocks, &protosrpc.BlockSubInfo{}, nil
	case SubHistoryBlocks:
		return r.subHistoryBlocks, &protosrpc.BlockSubInfo{}, nil
	case SubContractEvents:
		return r.subContractEvents, &protosrpc.ContractEventSubInfo{}, nil
	case SubHistoryEvents:
		return r.subHistoryEvents, &protosrpc.ContractEventSubInfo{}, nil
	case SubTransactionResult:
		return r.subTransactionResult, &protosrpc.TransactionSubInfo{}, nil
	case RPCMethodUnsubscribe:
		//如果需要传入context 需要将入参类型指定为Request
		return r.Unsubscribe, &protosrpc.UnsubscribeParam{}, nil
	default:
		return nil, nil, errors.New("method name does not exist")
	}
}

//GetBlockNumber 查看账本中最新块的编号
func (r *RPCMethodImpl) GetBlockNumber(ctx context.Context, in interface{}) (interface{}, int32, error) {
	r.module.Log.Debugf(LogCall, utils.FunName(), nil)
	//获取能查到的最大区块
	blockNum, err := r.RequestBlockNumber()
	if err != nil {
		return utils.Uint642Str(0), rpcerror.GetBlockNumberFailed.GetStatus(), err
	}
	return utils.Uint642Str(blockNum), rpcerror.Success.GetStatus(), nil
}

//RequestBlockNumber RequestBlockNumber
func (r *RPCMethodImpl) RequestBlockNumber() (uint64, error) {
	resp, err := utils.SendMqRequestWithResp(r.module.workerClient, types.BlockChainModule, mq.ModeWorker.String(), types.EventGetBlockNumber, nil, r.requestTimeOut)
	if err != nil || resp.Err != nil {
		return 0, err
	}

	//账本高度-1
	blockNum, ok := resp.Data.(uint64)
	if !ok {
		return 0, errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	return blockNum, nil
}

// GetBlockExecNumber 获取账本中执行区块高度
func (r *RPCMethodImpl) GetBlockExecNumber(ctx context.Context, in interface{}) (interface{}, int32, error) {
	r.module.Log.Debugf(LogCall, utils.FunName(), nil)
	//
	blockExecNum, err := r.RequestBlockExecNumber()
	if err != nil {
		return utils.Uint642Str(0), rpcerror.GetBlockExecNumberFailed.GetStatus(), err
	}
	return utils.Uint642Str(blockExecNum), rpcerror.Success.GetStatus(), nil
}

//RequestBlockExecNumber RequestBlockExecNumber
func (r *RPCMethodImpl) RequestBlockExecNumber() (uint64, error) {
	resp, err := utils.SendMqRequestWithResp(r.module.workerClient, types.BlockChainModule, mq.ModeWorker.String(), types.EventGetMaxExecutedBlockNumber, nil, r.requestTimeOut)
	if err != nil || resp.Err != nil {
		return 0, err
	}

	//账本高度-1
	blockNum, ok := resp.Data.(uint64)
	if !ok {
		return 0, errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	return blockNum, nil
}

//GetLatestExecutedBlock 获取最新区块
func (r *RPCMethodImpl) GetLatestExecutedBlock(ctx context.Context, in interface{}) (interface{}, int32, error) {
	r.module.Log.Debugf(LogCall, utils.FunName())

	resp, err := utils.SendMqRequestWithResp(r.module.workerClient, types.BlockChainModule, mq.ModeWorker.String(), types.EventGetLastBlock, nil, r.requestTimeOut)
	if err != nil {
		return nil, rpcerror.RequestMqMsgFailed.GetStatus(), err
	}

	eb, ok := resp.Data.(*protosledger.ExecutedBlock)
	if !ok {
		return nil, rpcerror.IllegalResponse.GetStatus(), errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	if eb == nil {
		return eb, rpcerror.NotExist.GetStatus(), fmt.Errorf("block not exist")
	}
	return eb, rpcerror.Success.GetStatus(), nil
}

//GetExecutedBlockByNumber 根据块号
func (r *RPCMethodImpl) GetExecutedBlockByNumber(ctx context.Context, in interface{}) (interface{}, int32, error) {
	var request = &protosrpc.BlockNumberParam{}
	_, err := GetParam(request, in)
	if err != nil {
		return nil, rpcerror.WrongParam.GetStatus(), err
	}

	r.module.Log.Debugf(LogCall, utils.FunName(), request)

	blockNumber, err := utils.Str2Uint64(request.BlockNumber)
	if err != nil {
		return nil, rpcerror.WrongParam.GetStatus(), errors.New(rpcerror.WrongParam.GetMessage())
	}

	var req = &protoscommon.ReqBlock{Height: blockNumber}
	resp, err := utils.SendMqRequestWithResp(r.module.workerClient, types.BlockChainModule, mq.ModeWorker.String(), types.EventGetExecutedBlock, req, r.requestTimeOut)
	if err != nil {
		return nil, rpcerror.RequestMqMsgFailed.GetStatus(), err
	}

	eb, ok := resp.Data.(*protosledger.ExecutedBlock)
	if !ok {
		return nil, rpcerror.IllegalResponse.GetStatus(), errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	return eb, rpcerror.Success.GetStatus(), nil
}

//GetExecutedBlocks 根据块号获取区块列表
func (r *RPCMethodImpl) GetExecutedBlocks(ctx context.Context, in interface{}) (interface{}, int32, error) {
	var request = &protosrpc.BlockNumberRangeParam{}
	_, err := GetParam(request, in)
	if err != nil {
		return nil, rpcerror.WrongParam.GetStatus(), err
	}

	r.module.Log.Debugf(LogCall, utils.FunName(), request)

	startBlockNum, errStart := utils.Str2Uint64(request.StartBlockNumber)
	endBlockNum, errEnd := utils.Str2Uint64(request.EndBlockNumber)
	if errStart != nil || errEnd != nil || startBlockNum > endBlockNum {
		return nil, rpcerror.WrongParam.GetStatus(), rpcerror.Errorf("startBlockNum or endBlockNum not illegal")
	}
	err = r.blockNumUint64Check(startBlockNum, endBlockNum, constant.GetBlockPageSize)
	if err != nil {
		return nil, rpcerror.WrongParam.GetStatus(), err
	}

	var req = &protoscommon.ReqBlocks{Start: startBlockNum, End: endBlockNum}
	resp, err := utils.SendMqRequestWithResp(r.module.workerClient, types.BlockChainModule, mq.ModeWorker.String(), types.EventGetExecutedBlocks, req, r.requestTimeOut)
	if err != nil {
		return nil, rpcerror.RequestMqMsgFailed.GetStatus(), err
	}

	ebs, ok := resp.Data.([]*protosledger.ExecutedBlock)
	if !ok {
		return nil, rpcerror.IllegalResponse.GetStatus(), errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	if len(ebs) == 0 {
		return nil, rpcerror.NotExist.GetStatus(), errors.New(rpcerror.NotExist.GetMessage())
	}
	return &protosrpc.ExecutedBlocks{Results: ebs}, rpcerror.Success.GetStatus(), nil

}

//GetBlockByNumber 根据块号查询block
func (r *RPCMethodImpl) GetBlockByNumber(ctx context.Context, in interface{}) (interface{}, int32, error) {
	var request = &protosrpc.BlockNumberParam{}
	_, err := GetParam(request, in)
	if err != nil {
		return nil, rpcerror.WrongParam.GetStatus(), err
	}

	r.module.Log.Debugf(LogCall, utils.FunName(), request)

	blockNumber, err := utils.Str2Uint64(request.BlockNumber)
	if err != nil {
		return nil, rpcerror.WrongParam.GetStatus(), errors.New(rpcerror.WrongParam.GetMessage())
	}

	var req = &protoscommon.ReqBlock{Height: blockNumber}
	resp, err := utils.SendMqRequestWithResp(r.module.workerClient, types.BlockChainModule, mq.ModeWorker.String(), types.EventGetBlock, req, r.requestTimeOut)
	if err != nil || resp.Err != nil {
		return nil, rpcerror.GetBlockByNumberFailed.GetStatus(), err
	}
	if resp.Data == nil {
		return nil, rpcerror.NotExist.GetStatus(), errors.New(rpcerror.NotExist.GetMessage())
	}

	blockResult, ok := resp.Data.(*protosledger.Block)
	if !ok {
		return nil, rpcerror.IllegalResponse.GetStatus(), errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	return blockResult, rpcerror.Success.GetStatus(), nil
}

//GetBalance 获取地址余额
func (r *RPCMethodImpl) GetBalance(ctx context.Context, in interface{}) (interface{}, int32, error) {
	var request = &protosrpc.GetBalanceParam{}
	_, err := GetParam(request, in)
	if err != nil {
		return nil, rpcerror.WrongParam.GetStatus(), err
	}
	errorCode := rpcerror.GetBalanceFailed.GetStatus()
	r.module.Log.Debugf(LogCall, utils.FunName(), request)
	reply, err := r.workerQueryMq(nativeTypes.NativeToken, nativeTypes.QueryBalanceMethod, request)
	if err != nil {
		return nil, errorCode, errors.New(rpcerror.GetBalanceFailed.FormatMsg(err.Error()))
	}
	if !reply.IsOk {
		if string(reply.Msg) == types.ErrNotFound.Error() {
			return utils.Uint642Str(0), rpcerror.Success.GetStatus(), nil
		}
		return nil, errorCode, errors.New(string(reply.Msg))
	}
	if reply.Msg == nil {
		return nil, rpcerror.NotExist.GetStatus(), errors.New(rpcerror.NotExist.GetMessage())
	}
	accountI, err := util.Bytes2Struct(&protoscommon.Account{}, reply.Msg)
	if err != nil {
		return nil, errorCode, errors.New(rpcerror.GetBalanceFailed.FormatMsg(err.Error()))
	}
	account, ok := accountI.(*protoscommon.Account)
	if !ok {
		return nil, rpcerror.IllegalResponse.GetStatus(), errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	return utils.Uint642Str(account.GetBalance()), rpcerror.Success.GetStatus(), nil
}

//Pure 方法调用worker
func (r *RPCMethodImpl) Pure(ctx context.Context, in interface{}) (interface{}, int32, error) {

	request, ok := in.(*protosrpc.Request)
	if !ok {
		return nil, rpcerror.WrongParam.GetStatus(), errors.New(rpcerror.WrongParam.GetMessage())
	}
	errorCode := rpcerror.PureMethodFailed.GetStatus()
	r.module.Log.Debugf(LogCall, utils.FunName(), request)
	if request == nil {
		return nil, rpcerror.WrongParam.GetStatus(), errors.New(rpcerror.WrongParam.GetMessage())
	}
	reply, err := r.workerQueryBytes(types2.JvmName, types2.QueryContractPureMethod, request.Params)
	if err != nil {
		return nil, errorCode, errors.New(rpcerror.PureMethodFailed.FormatMsg(err.Error()))
	}
	if !reply.IsOk {
		return nil, errorCode, errors.New(string(reply.Msg))
	}
	if reply.Msg == nil {
		return nil, rpcerror.NotExist.GetStatus(), errors.New(rpcerror.NotExist.GetMessage())
	}
	receiptI, err := util.Bytes2Struct(&protosledger.TransactionReceipt{}, reply.Msg)
	if err != nil {
		return nil, errorCode, errors.New(rpcerror.PureMethodFailed.FormatMsg(err.Error()))
	}
	receipt, ok := receiptI.(*protosledger.TransactionReceipt)
	if !ok {
		return nil, rpcerror.IllegalResponse.GetStatus(), errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	return receipt, rpcerror.Success.GetStatus(), nil
}

//GetContractVersion 获取合约版本
func (r *RPCMethodImpl) GetContractVersion(ctx context.Context, in interface{}) (interface{}, int32, error) {
	request, ok := in.(*protosrpc.Request)
	if !ok {
		return nil, rpcerror.WrongParam.GetStatus(), errors.New(rpcerror.WrongParam.GetMessage())
	}
	errorCode := rpcerror.GetContractVersionFailed.GetStatus()
	r.module.Log.Debugf(LogCall, utils.FunName(), request)
	if request == nil {
		return nil, errorCode, errors.New(rpcerror.WrongParam.GetMessage())
	}
	reply, err := r.workerQueryBytes(types2.JvmName, types2.QueryContractVersionMethod, request.Params)
	if err != nil {
		return nil, errorCode, errors.New(rpcerror.GetContractVersionFailed.FormatMsg(err.Error()))
	}
	if !reply.IsOk {
		return nil, errorCode, errors.New(string(reply.Msg))
	}
	if reply.Msg == nil {
		return nil, rpcerror.NotExist.GetStatus(), errors.New(rpcerror.NotExist.GetMessage())
	}
	receiptI, err := util.Bytes2Struct(&protoscommon.ContractVersion{}, reply.Msg)
	if err != nil {
		return nil, errorCode, errors.New(rpcerror.GetContractVersionFailed.FormatMsg(err.Error()))
	}
	receipt, ok := receiptI.(*protoscommon.ContractVersion)
	if !ok {
		return nil, rpcerror.IllegalResponse.GetStatus(), errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	return receipt, rpcerror.Success.GetStatus(), nil
}

//SendTransactionMonitor 长连接发送交易
func (r *RPCMethodImpl) SendTransactionMonitor(ctx context.Context) {

	for {
		select {
		case <-ctx.Done():
			r.module.Log.Infof("tx receive monitor closed")
			return
		case tx := <-r.addChan:
			err := r.send(tx)
			if err != nil {
				r.module.Log.Errorf("grpc stream send tx failed, reason:%v, reset p2p stream", err)
			}
		}
	}
}

func (r *RPCMethodImpl) send(tx *protosledger.Transaction) error {
	//如果是平行链
	var mainTopic string
	if r.module.module.IsParallel() {

		mainTopic = types.ParachainModule
	} else {
		mainTopic = types.P2pModule
	}
	if r.module.Log.IsTraceEnabled() {
		hash := tx.Hash()
		txLog := r.module.Log.WithField(logging.LogFieldTxHash, hash.GetShortString())
		txLog.Tracef("tx send to %s", mainTopic)
	}
	return utils.SendMqRequest(r.module.topicClient, mainTopic, mq.ModeTopic.String(), types.EventTxBroadcast, tx)
}

//GetParam 参数解析
func GetParam(res interface{}, in interface{}) (interface{}, error) {
	req, ok := in.(*protosrpc.Request)
	if !ok {
		return nil, errors.New(rpcerror.WrongParam.GetMessage())
	}
	result, err := util.Bytes2Struct(res, req.Params)
	if err != nil {
		return nil, errors.New(rpcerror.WrongParam.GetMessage())
	}
	return result, nil
}

//SendSignedTransactionAsyn 异步发送已签名交易,转发给p2p
func (r *RPCMethodImpl) SendSignedTransactionAsyn(ctx context.Context, in interface{}) (interface{}, int32, error) {
	var request = &protosledger.Transaction{}
	_, err := GetParam(request, in)
	if err != nil {
		return nil, rpcerror.WrongParam.GetStatus(), err
	}
	return r.SendSignedTransaction(request)
}

//SendSignedTransaction 发送交易
func (r *RPCMethodImpl) SendSignedTransaction(request *protosledger.Transaction) (interface{}, int32, error) {
	hash := request.Hash()

	//判断交易大小
	buf, _ := util.Struct2Bytes(request)
	length := len(buf)
	if length > int(r.GenesisConf.BlockLimit.GetMaxTxSizeBytes()) {
		txLog := r.module.Log.WithField(logging.LogFieldTxHash, hash.GetShortString())
		return hash, rpcerror.TxSendFailed.GetStatus(), rpcerror.LoggerErrorf(txLog, "tx exceed limit")
	}
	if request.Fee < r.GenesisConf.GetFeeLimit().GetMinTxFee() {
		return hash, rpcerror.TxSendFailed.GetStatus(), rpcerror.LoggerErrorf(r.module.Log.WithField(logging.LogFieldTxHash, hash.GetShortString()), "tx fee small than minTxFee")
	}
	r.addChan <- request
	return hash, rpcerror.Success.GetStatus(), nil
}

//GetTransactionByHash 根据hash获取交易
func (r *RPCMethodImpl) GetTransactionByHash(ctx context.Context, in interface{}) (interface{}, int32, error) {
	var request = &protosrpc.TxHashParam{}
	_, err := GetParam(request, in)
	if err != nil {
		return nil, rpcerror.WrongParam.GetStatus(), err
	}
	buf, err := hex.DecodeString(request.TxHash)
	if err != nil {
		return nil, rpcerror.SerializeFail.GetStatus(), errors.New(rpcerror.SerializeFail.GetMessage())
	}
	var param = &protoscommon.ReqHashes{
		Hashes: [][]byte{buf},
	}
	r.module.Log.Debugf(LogCall, utils.FunName(), in.(*protosrpc.Request))

	resp, err := utils.SendMqRequestWithResp(r.module.workerClient, types.BlockChainModule, mq.ModeWorker.String(), types.EventTxQuery, param, r.requestTimeOut)
	if err != nil || resp.Err != nil {
		return nil, rpcerror.RequestMqMsgFailed.GetStatus(), err
	}
	txResults, ok := resp.Data.([]*protosledger.ExecutedTransaction)
	if !ok {
		return nil, rpcerror.IllegalResponse.GetStatus(), errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	if len(txResults) == 0 {
		return nil, rpcerror.NotExist.GetStatus(), errors.New(rpcerror.NotExist.GetMessage())
	}
	return txResults[0], rpcerror.Success.GetStatus(), nil
}

//GetPeerCount 查看当前节点连接的其他节点数
func (r *RPCMethodImpl) GetPeerCount(ctx context.Context, request interface{}) (interface{}, int32, error) {
	resp, err := utils.SendMqRequestWithResp(r.module.workerClient, types.P2pModule, mq.ModeWorker.String(), types.EventPeerInfo, nil, r.requestTimeOut)
	if err != nil || resp.Err != nil {
		return nil, rpcerror.GetPeerCountFailed.GetStatus(), rpcerror.Errorf("get peer count failed, restart rpc client, reason:%v", err)
	}
	if resp.Data == nil {
		return nil, rpcerror.NotExist.GetStatus(), errors.New(rpcerror.NotExist.GetMessage())
	}
	result, ok := resp.Data.(*protosp2p.PeerInfos)
	if !ok {
		return nil, rpcerror.IllegalResponse.GetStatus(), errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	return fmt.Sprint(len(result.Peers)), rpcerror.Success.GetStatus(), nil
}

//交易订阅
func (r *RPCMethodImpl) subTransactionResult(param *protosrpc.Request, stream protosrpc.Rpc_ProcessSubRequestServer) error {
	if param == nil {
		_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
			{Status: rpcerror.WrongParam.GetStatus(),
				Message: rpcerror.WrongParam.GetMessage(),
				SubType: protosrpc.Result_TXRESULT,
				SubMark: protosrpc.Result_REALTIME,
			}}})
		return rpcerror.Errorf("method SubTransactionResult has wrong parameters")
	}
	subInfoI, err := util.Bytes2Struct(&protosrpc.TransactionSubInfo{}, param.GetParams())
	subInfo, ok := subInfoI.(*protosrpc.TransactionSubInfo)
	if err != nil || !ok {
		_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
			{Status: rpcerror.SerializeFail.GetStatus(),
				Message: rpcerror.SerializeFail.GetMessage(),
				SubType: protosrpc.Result_TXRESULT,
				SubId:   subInfo.SubId,
				SubMark: protosrpc.Result_REALTIME}}})
		return rpcerror.Errorf("%s", err)
	}
	//更新时间戳
	subInfo.Timestamp = param.Timestamp
	ip, err := utils.GetClientIP(stream.Context())
	if err != nil {
		return rpcerror.Errorf("%s", err)
	}
	//打印收到新区块订阅的日志
	r.module.Log.WithField(logging.LogFieldTxHash, hex.EncodeToString(subInfo.TxHash)).WithField(logging.LogFieldP2pSubID, subInfo.GetSubId()).Debugf("get a transaction subscribe from %s", ip)
	if monitor, ok := r.module.endpointInstance.Get(ip); ok {
		if len(monitor.requestMap) >= constant.SubscribeNumberLimit {
			_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
				{Status: rpcerror.FlowControlFailed.GetStatus(),
					Message: rpcerror.FlowControlFailed.GetMessage(),
					SubType: protosrpc.Result_TXRESULT,
					SubId:   subInfo.SubId,
					SubMark: protosrpc.Result_REALTIME,
				}}})
			return errors.New(rpcerror.FlowControlFailed.GetMessage())
		}
		monitor.RequestMapPut(subInfo.SubId, subInfo)
	}
	return nil
}

//Unsubscribe 取消订阅
func (r *RPCMethodImpl) Unsubscribe(param *protosrpc.Request, stream protosrpc.Rpc_ProcessSubRequestServer) error {
	ip, err := utils.GetClientIP(stream.Context())
	if err != nil {
		return err
	}
	subID, err := util.Bytes2Struct(&protosrpc.UnsubscribeParam{}, param.GetParams())
	if err != nil {
		_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
			{Status: rpcerror.SerializeFail.GetStatus(),
				Message: rpcerror.SerializeFail.GetMessage(),
				SubType: protosrpc.Result_UNSUBSCRIBE,
				SubMark: protosrpc.Result_REALTIME}}})
		return rpcerror.Errorf("%s", err)
	}
	if monitor, ok := r.module.endpointInstance.Get(ip); ok {
		monitor.RequestMapDelete(subID.(*protosrpc.UnsubscribeParam).CancelSubId)
	}
	r.module.Log.Infof("unsubscribe subID:%s", subID.(*protosrpc.UnsubscribeParam).CancelSubId)
	_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
		{Status: rpcerror.Success.GetStatus(),
			SubId:   subID.(*protosrpc.UnsubscribeParam).SubId,
			Message: rpcerror.Success.GetMessage(),
			SubType: protosrpc.Result_UNSUBSCRIBE,
			SubMark: protosrpc.Result_REALTIME}}})
	return nil
}

func (r *RPCMethodImpl) subBlocks(param *protosrpc.Request, stream protosrpc.Rpc_ProcessSubRequestServer) error {
	if param == nil {
		_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
			{Status: rpcerror.WrongParam.GetStatus(),
				Message: "param is nil",
				SubType: protosrpc.Result_BLOCK,
				SubMark: protosrpc.Result_REALTIME,
			}}})
		return rpcerror.Errorf("method SubBlocks has wrong parameters")
	}
	subInfoI, err := util.Bytes2Struct(&protosrpc.BlockSubInfo{}, param.GetParams())
	subInfo, ok := subInfoI.(*protosrpc.BlockSubInfo)
	if err != nil || !ok {
		_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
			{Status: rpcerror.SerializeFail.GetStatus(),
				Message: "subscribe info serialize failed",
				SubType: protosrpc.Result_BLOCK,
				SubId:   subInfo.SubId,
				SubMark: protosrpc.Result_REALTIME,
			}}})
		return rpcerror.Errorf("%s", err)
	}
	ip, err := utils.GetClientIP(stream.Context())
	if err != nil {
		return rpcerror.Errorf("%s", err)
	}
	//打印收到新区块订阅的日志
	r.module.Log.WithField(logging.LogFieldP2pSubID, subInfo.GetSubId()).Infof("get a block subscribe from %s", ip)
	if monitor, ok := r.module.endpointInstance.Get(ip); ok {
		if len(monitor.requestMap) >= constant.SubscribeNumberLimit {
			_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
				{Status: rpcerror.FlowControlFailed.GetStatus(),
					Message: rpcerror.FlowControlFailed.GetMessage(),
					SubType: protosrpc.Result_BLOCK,
					SubId:   subInfo.SubId,
					SubMark: protosrpc.Result_REALTIME,
				}}})
			return errors.New(rpcerror.FlowControlFailed.GetMessage())
		}
		monitor.RequestMapPut(subInfo.SubId, subInfo)
	}
	return nil
}

//区间订阅区块
func (r *RPCMethodImpl) subHistoryBlocks(param *protosrpc.Request, stream protosrpc.Rpc_ProcessSubRequestServer) error {

	if param == nil {
		_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
			{Status: rpcerror.WrongParam.GetStatus(),
				Message: "param is nil",
				SubType: protosrpc.Result_BLOCK,
				SubMark: protosrpc.Result_HISTORY,
			}}})
		return rpcerror.Errorf("method SubBlocks has wrong parameters")
	}
	//struct2bean
	subInfoI, err := util.Bytes2Struct(&protosrpc.BlockSubInfo{}, param.GetParams())
	subInfo, ok := subInfoI.(*protosrpc.BlockSubInfo)
	if err != nil || !ok {
		_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
			{Status: rpcerror.SerializeFail.GetStatus(),
				Message: "subscribe info serialize failed",
				SubType: protosrpc.Result_BLOCK,
				SubId:   subInfo.SubId,
				SubMark: protosrpc.Result_HISTORY,
			}}})
		return rpcerror.Errorf("%s", err)
	}

	//参数校验
	if err := r.blockNumCheck(subInfo, constant.GetBlockPageSize); err != nil {
		_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
			{Status: rpcerror.WrongParam.GetStatus(),
				Message: err.Error(),
				SubType: protosrpc.Result_BLOCK,
				SubId:   subInfo.SubId,
				SubMark: protosrpc.Result_HISTORY,
			}}})
		return rpcerror.Errorf("query param not match")
	}

	ip, err := utils.GetClientIP(stream.Context())
	if err != nil {
		return rpcerror.Errorf("%s", err)
	}

	r.module.Log.WithField(logging.LogFieldP2pSubID, subInfo.GetSubId()).Infof("get a history block subscribe from %s", ip)

	//获取能查到的最大区块
	blockNum, err := r.RequestBlockNumber()
	if err != nil {
		_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
			{Status: rpcerror.GetBlockNumberFailed.GetStatus(),
				Message: err.Error(),
				SubType: protosrpc.Result_BLOCK,
				SubId:   subInfo.SubId,
				SubMark: protosrpc.Result_HISTORY,
			}}})
		return rpcerror.Errorf("%s", err)
	}
	var queryBlockNum uint64

	//字符串转uint64
	startBlockNumber, err1 := utils.Str2Uint64(subInfo.StartBlockNum)
	endBlockNumber, err2 := utils.Str2Uint64(subInfo.EndBlockNum)
	if err1 != nil || err2 != nil {
		_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
			{Status: rpcerror.WrongParam.GetStatus(),
				Message: "convert block number from string to uint64 failed",
				SubType: protosrpc.Result_BLOCK,
				SubId:   subInfo.SubId,
				SubMark: protosrpc.Result_HISTORY,
			}}})
		return rpcerror.Errorf("%s", err)
	}

	r.module.Log.Infof("ledgerNextBlockNum:%d, subInfo.StartBlockNum:%d, subInfo.EndBlockNum:%d", blockNum+1, startBlockNumber, endBlockNumber)
	if blockNum >= endBlockNumber {
		queryBlockNum = endBlockNumber
	} else {
		queryBlockNum = blockNum
	}

	//查询历史区块
	var req = &protoscommon.ReqBlocks{Start: startBlockNumber, End: queryBlockNum}
	resp, err := utils.SendMqRequestWithResp(r.module.workerClient, types.BlockChainModule, mq.ModeWorker.String(), types.EventGetExecutedBlocks, req, r.requestTimeOut)
	if err != nil {
		r.module.Log.Errorf("Get executed blocks failed, err: %s", err)
		err = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{{
			SubType:           protosrpc.Result_BLOCK,
			SubId:             subInfo.SubId,
			SubMark:           protosrpc.Result_HISTORY,
			LedgerBlockHeight: blockNum + 1,
			BlockNum:          queryBlockNum,
			Status:            rpcerror.Success.GetStatus(),
			Message:           rpcerror.Success.GetMessage(),
		}}})
		return nil
	}

	//获取results
	blocks, ok := resp.Data.([]*protosledger.ExecutedBlock)
	if !ok {
		r.module.Log.Errorf("Get executed blocks failed, err: %s", errors.New(rpcerror.IllegalResponse.GetMessage()))
	}
	var results []*protosrpc.Result
	for _, block := range blocks {
		_, _, val := subInfo.Filter(block, protosrpc.Result_HISTORY)
		results = append(results, val...)
	}

	//拼装总高度和submark
	for _, result := range results {
		result.LedgerBlockHeight = blockNum + 1
		result.SubMark = protosrpc.Result_HISTORY
	}

	//如果result为空，拼入空result
	if len(results) == 0 {
		results = append(results, &protosrpc.Result{
			SubType:           protosrpc.Result_BLOCK,
			SubId:             subInfo.SubId,
			SubMark:           protosrpc.Result_HISTORY,
			LedgerBlockHeight: blockNum + 1,
			BlockNum:          queryBlockNum,
			Status:            rpcerror.Success.GetStatus(),
			Message:           rpcerror.Success.GetMessage(),
		})
	}

	r.module.Log.Infof("subscribe blocks num:%d", len(blocks))
	err = stream.Send(&protosrpc.SubscriptionResult{Result: results})
	if err != nil {
		r.module.Log.Errorf("subscribe blocks message send failed, ip:%s, reason:%v", ip, err)
	}
	return nil
}

//区间订阅合约事件
func (r *RPCMethodImpl) subContractEvents(param *protosrpc.Request, stream protosrpc.Rpc_ProcessSubRequestServer) error {
	if param == nil {
		_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
			{Status: rpcerror.WrongParam.GetStatus(),
				Message: "param is nil",
				SubType: protosrpc.Result_CONTRACTEVENT,
				SubMark: protosrpc.Result_REALTIME,
			}}})
		return rpcerror.Errorf("method SubContractEvent has wrong parameters")
	}
	subInfoI, err := util.Bytes2Struct(&protosrpc.ContractEventSubInfo{}, param.GetParams())
	subInfo, ok := subInfoI.(*protosrpc.ContractEventSubInfo)

	if err != nil || !ok {
		_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
			{Status: rpcerror.SerializeFail.GetStatus(),
				Message: "subscribe info serialize failed",
				SubType: protosrpc.Result_CONTRACTEVENT,
				SubId:   subInfo.SubId,
				SubMark: protosrpc.Result_REALTIME,
			}}})
		return rpcerror.Errorf("%s", err)
	}

	//检验合约订阅参数
	if err = r.contractSubInfoCheck(subInfo); err != nil {
		_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
			{Status: rpcerror.WrongParam.GetStatus(),
				Message: err.Error(),
				SubType: protosrpc.Result_CONTRACTEVENT,
				SubId:   subInfo.SubId,
				SubMark: protosrpc.Result_REALTIME,
			}}})
		return rpcerror.Errorf("%s", err)
	}
	ip, err := utils.GetClientIP(stream.Context())
	if err != nil {
		return rpcerror.Errorf("%s", err)
	}
	//打印收到新区块订阅的日志
	r.module.Log.WithField(logging.LogFieldP2pSubID, subInfo.GetSubId()).Infof("get a contract subscribe from %s", ip)
	if monitor, ok := r.module.endpointInstance.Get(ip); ok {
		if len(monitor.requestMap) >= constant.SubscribeNumberLimit {
			_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
				{Status: rpcerror.FlowControlFailed.GetStatus(),
					Message: rpcerror.FlowControlFailed.GetMessage(),
					SubType: protosrpc.Result_CONTRACTEVENT,
					SubId:   subInfo.SubId,
					SubMark: protosrpc.Result_REALTIME,
				}}})
			return errors.New(rpcerror.FlowControlFailed.GetMessage())
		}
		monitor.RequestMapPut(subInfo.SubId, subInfo)
	}
	return nil
}

func (r *RPCMethodImpl) subHistoryEvents(param *protosrpc.Request, stream protosrpc.Rpc_ProcessSubRequestServer) error {
	if param == nil {
		_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
			{Status: rpcerror.WrongParam.GetStatus(),
				Message: "request is nil",
				SubType: protosrpc.Result_CONTRACTEVENT,
				SubMark: protosrpc.Result_HISTORY,
			}}})
		return rpcerror.Errorf("method SubBlocks has wrong parameters")
	}
	//struct2bean
	subInfoI, err := util.Bytes2Struct(&protosrpc.ContractEventSubInfo{}, param.GetParams())
	subInfo, ok := subInfoI.(*protosrpc.ContractEventSubInfo)
	if err != nil || !ok {
		_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
			{Status: rpcerror.SerializeFail.GetStatus(),
				Message: "subscribe info deserialize failed",
				SubType: protosrpc.Result_CONTRACTEVENT,
				SubId:   subInfo.SubId,
				SubMark: protosrpc.Result_HISTORY,
			}}})
		return rpcerror.Errorf("%s", err)
	}
	//检验合约订阅参数
	if err = r.contractSubInfoCheck(subInfo); err != nil {
		_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
			{Status: rpcerror.WrongParam.GetStatus(),
				Message: err.Error(),
				SubType: protosrpc.Result_CONTRACTEVENT,
				SubId:   subInfo.SubId,
				SubMark: protosrpc.Result_HISTORY,
			}}})
		return rpcerror.Errorf("%s", err)
	}

	//参数校验
	if err := r.blockNumCheck(subInfo, constant.GetEventPageSize); err != nil {
		_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
			{Status: rpcerror.WrongParam.GetStatus(),
				Message: err.Error(),
				SubType: protosrpc.Result_CONTRACTEVENT,
				SubId:   subInfo.SubId,
				SubMark: protosrpc.Result_HISTORY,
			}}})
		return rpcerror.Errorf("query param not match")
	}

	ip, err := utils.GetClientIP(stream.Context())
	if err != nil {
		return rpcerror.Errorf("get stream ip failed, reason:%v", err)
	}

	r.module.Log.WithField(logging.LogFieldP2pSubID, subInfo.GetSubId()).Infof("get a history event subscribe from %s", ip)

	//字符串转uint64
	startBlockNumber, err1 := utils.Str2Uint64(subInfo.StartBlockNum)
	endBlockNumber, err2 := utils.Str2Uint64(subInfo.EndBlockNum)
	if err1 != nil || err2 != nil {
		_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
			{Status: rpcerror.WrongParam.GetStatus(),
				Message: "convert block number from string to uint64 failed",
				SubType: protosrpc.Result_CONTRACTEVENT,
				SubId:   subInfo.SubId,
				SubMark: protosrpc.Result_HISTORY,
			}}})
		return rpcerror.Errorf("%s", err)
	}

	//获取能查到的最大区块
	blockNum, err := r.RequestBlockNumber()
	if err != nil {
		_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
			{Status: rpcerror.GetBlockNumberFailed.GetStatus(),
				Message: rpcerror.GetBlockNumberFailed.GetMessage(),
				SubType: protosrpc.Result_CONTRACTEVENT,
				SubId:   subInfo.SubId,
				SubMark: protosrpc.Result_HISTORY,
			}}})
		return rpcerror.Errorf("%s", err)
	}
	var queryBlockNum uint64
	if blockNum >= endBlockNumber {
		queryBlockNum = endBlockNumber
	} else {
		queryBlockNum = blockNum
	}

	events, _ := r.getContractByBlock(startBlockNumber, queryBlockNum, subInfo.ContractId.Name, subInfo.ContractId.Version, subInfo.Name, subInfo.IndexedParamFirst, subInfo.IndexedParamSecond, subInfo.IndexedParamThird)

	//获取results
	var results []*protosrpc.Result

	for _, event := range events {
		buffer, err := util.Struct2Bytes(event)
		if err != nil {
			_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
				{Status: rpcerror.SerializeFail.GetStatus(),
					Message: "event serialize failed",
					SubType: protosrpc.Result_CONTRACTEVENT,
					SubId:   subInfo.SubId,
					SubMark: protosrpc.Result_HISTORY,
				}}})
			return rpcerror.Errorf("%s", err)
		}
		result := &protosrpc.Result{
			SubType:           protosrpc.Result_CONTRACTEVENT,
			SubId:             subInfo.SubId,
			SubMark:           protosrpc.Result_HISTORY,
			LedgerBlockHeight: blockNum + 1,
			Status:            rpcerror.Success.GetStatus(),
			Message:           rpcerror.Success.GetMessage(),
			Payload:           buffer,
			BlockNum:          event.BlockNumber,
		}
		results = append(results, result)
	}

	//如果result为空，拼入空result
	if len(results) == 0 {
		results = append(results, &protosrpc.Result{
			SubType:           protosrpc.Result_CONTRACTEVENT,
			SubId:             subInfo.SubId,
			SubMark:           protosrpc.Result_HISTORY,
			LedgerBlockHeight: blockNum + 1,
			BlockNum:          queryBlockNum,
			Status:            rpcerror.Success.GetStatus(),
			Message:           rpcerror.Success.GetMessage(),
		})
	}

	r.module.Log.Infof("subscribe events num:%d", len(events))
	err = stream.Send(&protosrpc.SubscriptionResult{Result: results})
	if err != nil {
		r.module.Log.Errorf("subscribe blocks message send failed, ip:%s, reason:%v", ip, err)
	}
	return nil
}

func (r *RPCMethodImpl) contractSubInfoCheck(param *protosrpc.ContractEventSubInfo) error {
	if param.ContractId == nil || param.Name == "" {
		return rpcerror.Errorf("contract subscribe info illegal, contractId:%v, eventName:%v, contractAddress:%s", param.ContractId, param.Name, param.ContractAddress)
	}
	return nil
}

func (r *RPCMethodImpl) blockNumUint64Check(startBlockNumber uint64, endBlockNumber uint64, limit uint64) error {
	//字符串转uint64
	if endBlockNumber-startBlockNumber >= limit {
		return rpcerror.Errorf("blockNumber exceed limit, startBlockNum:%d, endBlockNum:%d, limit:%d", startBlockNumber, endBlockNumber, limit)
	}
	return nil
}

func (r *RPCMethodImpl) blockNumCheck(param protosrpc.IBlockSubscribe, limit uint64) error {
	//字符串转uint64
	startBlockNumber, err1 := utils.Str2Uint64(param.GetStartBlockNum())
	endBlockNumber, err2 := utils.Str2Uint64(param.GetEndBlockNum())
	if err1 != nil || err2 != nil || startBlockNumber > endBlockNumber {
		return rpcerror.Errorf("blockNumber illegal, startBlockNum:%s, endBlockNum:%s", param.GetStartBlockNum(), param.GetEndBlockNum())
	}

	if endBlockNumber-startBlockNumber >= limit {
		return rpcerror.Errorf("blockNumber exceed limit, startBlockNum:%s, endBlockNum:%s, limit:%d", param.GetStartBlockNum(), param.GetEndBlockNum(), limit)
	}
	return nil
}

func (r *RPCMethodImpl) getContractByBlock(startBlockNum uint64, endBlockNum uint64, contractName string, contractVersion string, eventName string, index1 []byte, index2 []byte, index3 []byte) ([]*protoscommon.ContractEvent, error) {
	var eventInfo = &protosrpc.ContractEventSubInfo{
		ContractId:         &protoscommon.ContractIdentifier{Name: contractName, Version: contractVersion},
		StartBlockNum:      strconv.FormatUint(startBlockNum, 10),
		EndBlockNum:        strconv.FormatUint(endBlockNum, 10),
		Name:               eventName,
		IndexedParamFirst:  index1,
		IndexedParamSecond: index2,
		IndexedParamThird:  index3,
	}
	buf, err := proto.Marshal(eventInfo)
	if err != nil {
		return nil, err
	}
	reply, err := r.workerQueryBytes(types2.JvmName, types2.QueryHistoryEventMethod, buf)
	if err != nil {
		return nil, err
	}
	if !reply.IsOk {
		return nil, errors.New(string(reply.Msg))
	}
	if reply.Msg == nil {
		return nil, errors.New(rpcerror.NotExist.GetMessage())
	}
	receiptI, err := util.Bytes2Struct(&protosrpc.ContractEvents{}, reply.Msg)
	if err != nil {
		return nil, errors.New(rpcerror.SerializeFail.GetMessage())
	}
	receipt, ok := receiptI.(*protosrpc.ContractEvents)
	if !ok {
		return nil, errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	return receipt.Results, nil
}

//GetExecutedBlock 获取执行后的区块
func GetExecutedBlock(client mq.Client, blockNum uint64) (*protosledger.ExecutedBlock, error) {
	var req = &protoscommon.ReqBlock{Height: blockNum}
	resp, err := utils.SendMqRequestWithResp(client, types.BlockChainModule, mq.ModeWorker.String(), types.EventGetExecutedBlock, req, time.Second*constant.MqTimeOut)
	if err != nil {
		return nil, rpcerror.Errorf("get executed block failed, reason:%v", err)
	}

	block, ok := resp.Data.(*protosledger.ExecutedBlock)
	if !ok {
		return nil, errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	return block, nil
}

//GetBlockExecResults 获取区块执行结果
func (r *RPCMethodImpl) GetBlockExecResults(ctx context.Context, in interface{}) (interface{}, int32, error) {
	var request = &protosrpc.BlockNumberRangeParam{}
	_, err := GetParam(request, in)
	if err != nil {
		return nil, rpcerror.WrongParam.GetStatus(), err
	}

	r.module.Log.Debugf(LogCall, utils.FunName(), request)

	startBlockNum, errStart := utils.Str2Uint64(request.StartBlockNumber)
	endBlockNum, errEnd := utils.Str2Uint64(request.EndBlockNumber)
	if errStart != nil || errEnd != nil || startBlockNum > endBlockNum {
		return nil, rpcerror.WrongParam.GetStatus(), rpcerror.Errorf("startBlockNum or endBlockNum not illegal")
	}
	err = r.blockNumUint64Check(startBlockNum, endBlockNum, constant.GetBlockPageSize)
	if err != nil {
		return nil, rpcerror.WrongParam.GetStatus(), err
	}
	var req = &protoscommon.ReqBlocks{Start: startBlockNum, End: endBlockNum}
	resp, err := utils.SendMqRequestWithResp(r.module.workerClient, types.BlockChainModule, mq.ModeWorker.String(), types.EventGetExecutedBlocks, req, r.requestTimeOut)
	if err != nil {
		return nil, rpcerror.GetBlockByNumberFailed.GetStatus(), err
	}
	execBlks, ok := resp.Data.([]*protosledger.ExecutedBlock)
	if !ok {
		return nil, rpcerror.IllegalResponse.GetStatus(), errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	if len(execBlks) == 0 {
		return nil, rpcerror.NotExist.GetStatus(), errors.New(rpcerror.NotExist.GetMessage())
	}
	var res []*protosledger.BlockExecResult
	for _, item := range execBlks {
		res = append(res, item.BlockExecResult)
	}
	return &protosrpc.BlockExecResults{Results: res}, rpcerror.Success.GetStatus(), nil
}

//GetHistoryContractEvents 获取历史合约事件
func (r *RPCMethodImpl) GetHistoryContractEvents(ctx context.Context, in interface{}) (interface{}, int32, error) {
	var request = &protosrpc.ContractEventSubInfo{}
	_, err := GetParam(request, in)
	if err != nil {
		return nil, rpcerror.WrongParam.GetStatus(), err
	}

	err = r.contractSubInfoCheck(request)
	if err != nil {
		return nil, rpcerror.WrongParam.GetStatus(), err
	}
	startBlockNumber, err1 := utils.Str2Uint64(request.StartBlockNum)
	endBlockNumber, err2 := utils.Str2Uint64(request.EndBlockNum)
	if err1 != nil || err2 != nil {
		return nil, rpcerror.WrongParam.GetStatus(),
			fmt.Errorf("startBlockNumber or endBlockNumber not illegal, startBlockNum:%s, endBlockNum:%s", request.StartBlockNum, request.EndBlockNum)
	}
	err = r.blockNumUint64Check(startBlockNumber, endBlockNumber, constant.GetEventPageSize)
	if err != nil {
		return nil, rpcerror.WrongParam.GetStatus(), err
	}
	events, err := r.getContractByBlock(startBlockNumber, endBlockNumber, request.ContractId.Name, request.ContractId.Version, request.Name, request.IndexedParamFirst, request.IndexedParamSecond, request.IndexedParamThird)
	if err != nil {
		return nil, rpcerror.CommonFail.GetStatus(), err
	}
	if len(events) == 0 {
		return nil, rpcerror.NotExist.GetStatus(), errors.New(rpcerror.NotExist.GetMessage())
	}
	return &protosrpc.ContractEvents{Results: events}, rpcerror.Success.GetStatus(), nil
}

//GetPeerInfos 获取节点信息
func (r *RPCMethodImpl) GetPeerInfos(ctx context.Context, in interface{}) (interface{}, int32, error) {
	resp, err := utils.SendMqRequestWithResp(r.module.workerClient, types.P2pModule, mq.ModeWorker.String(), types.EventPeerInfo, nil, r.requestTimeOut)
	if err != nil || resp.Err != nil {
		return nil, rpcerror.GetPeerInfoFailed.GetStatus(), rpcerror.Errorf("get peer info failed, reason:%v", err)
	}
	result, ok := resp.Data.(*protosp2p.PeerInfos)
	if !ok {
		return nil, rpcerror.IllegalResponse.GetStatus(), errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	return result, rpcerror.Success.GetStatus(), nil
}

// GetConsensusInfo 查询共识相关信息
func (r *RPCMethodImpl) GetConsensusInfo(ctx context.Context, in interface{}) (interface{}, int32, error) {
	resp, err := utils.SendMqRequestWithResp(r.module.workerClient, types.ConsensusModule, mq.ModeWorker.String(), types.EventConsensusQuery, nil, r.requestTimeOut)
	if err != nil || resp.Err != nil {
		return nil, rpcerror.GetPeerInfoFailed.GetStatus(), rpcerror.Errorf("get peer info failed, reason:%v", err)
	}
	result, ok := resp.Data.(*pbconfig.ConsensusParam)
	if !ok {
		return nil, rpcerror.IllegalResponse.GetStatus(), errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	return result, rpcerror.Success.GetStatus(), nil
}

//IsSynced 节点高度是否同步
func (r *RPCMethodImpl) IsSynced(ctx context.Context, in interface{}) (interface{}, int32, error) {
	resp, err := utils.SendMqRequestWithResp(r.module.workerClient, types.BlockChainModule, mq.ModeWorker.String(), types.EventIsSynched, nil, r.requestTimeOut)
	if err != nil || resp.Err != nil {
		return false, rpcerror.GetIsSyncedFailed.GetStatus(), rpcerror.Errorf("get peer is synced failed, reason:%v", err)
	}
	result, ok := resp.Data.(bool)
	if !ok {
		return false, rpcerror.IllegalResponse.GetStatus(), errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	return result, rpcerror.Success.GetStatus(), nil
}

//GetP2pPeerID 获取p2p peerID
func (r *RPCMethodImpl) GetP2pPeerID(ctx context.Context, in interface{}) (interface{}, int32, error) {
	resp, err := utils.SendMqRequestWithResp(r.module.workerClient, types.P2pModule, mq.ModeWorker.String(), types.EventPeerID, nil, r.requestTimeOut)
	if err != nil || resp.Err != nil {
		return "", rpcerror.GetPeerIDFailed.GetStatus(), rpcerror.Errorf("get peer id failed, reason:%v", err)
	}

	result, ok := resp.Data.(string)
	if !ok {
		return nil, rpcerror.IllegalResponse.GetStatus(), errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	return result, rpcerror.Success.GetStatus(), nil
}

//PeekTxPool peek tx in tx pool
func (r *RPCMethodImpl) PeekTxPool(ctx context.Context, in interface{}) (interface{}, int32, error) {
	var request = &protosrpc.TxHashParam{}
	_, err := GetParam(request, in)
	if err != nil {
		return nil, rpcerror.WrongParam.GetStatus(), err
	}
	buf, err := hex.DecodeString(request.TxHash)
	if err != nil {
		return nil, rpcerror.SerializeFail.GetStatus(), errors.New(rpcerror.SerializeFail.GetMessage())
	}
	resp, err := utils.SendMqRequestWithResp(r.module.workerClient, types.P2pModule, mq.ModeWorker.String(), types.EventTxPoolPeekTx, protos.Hash(buf), r.requestTimeOut)
	if err != nil || resp.Err != nil {
		if resp.Err == p2perror.ErrNotFound {
			return nil, rpcerror.Success.GetStatus(), nil
		}
		return nil, rpcerror.PeekPendingTxFailed.GetStatus(), rpcerror.Errorf("peek txpool failed, reason:%v", err)
	}
	result, ok := resp.Data.(*protosledger.Transaction)
	if !ok {
		return nil, rpcerror.IllegalResponse.GetStatus(), errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	return result, rpcerror.Success.GetStatus(), nil
}

//TxPoolSize get txpool size
func (r *RPCMethodImpl) TxPoolSize(ctx context.Context, in interface{}) (interface{}, int32, error) {
	resp, err := utils.SendMqRequestWithResp(r.module.workerClient, types.P2pModule, mq.ModeWorker.String(), types.EventTxPoolSize, nil, r.requestTimeOut)
	if err != nil || resp.Err != nil {
		return nil, rpcerror.TxPoolSizeFailed.GetStatus(), rpcerror.Errorf("get txpool size failed, reason:%v", err)
	}
	result, ok := resp.Data.(int)
	if !ok {
		return nil, rpcerror.IllegalResponse.GetStatus(), errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	return strconv.Itoa(result), rpcerror.Success.GetStatus(), nil
}

//TxPoolHashes query hashes of transactions that prepare pending
func (r *RPCMethodImpl) TxPoolHashes(ctx context.Context, in interface{}) (interface{}, int32, error) {
	var request uint16
	param, err := GetParam(request, in)
	if err != nil {
		return nil, rpcerror.WrongParam.GetStatus(), err
	}
	resp, err := utils.SendMqRequestWithResp(r.module.workerClient, types.P2pModule, mq.ModeWorker.String(), types.EventTxPoolHashes, param, r.requestTimeOut)
	if err != nil || resp.Err != nil {
		return nil, rpcerror.TxPoolHashesFailed.GetStatus(), rpcerror.Errorf("get transaction hashes failed, reason:%v", err)
	}
	hashes, ok := resp.Data.([]string)
	if !ok {
		return nil, rpcerror.IllegalResponse.GetStatus(), errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	return &protosrpc.Strings{
		Results: hashes,
	}, rpcerror.Success.GetStatus(), nil
}

// QueryContract  query contract
func (r *RPCMethodImpl) QueryContract(ctx context.Context, in interface{}) (interface{}, int32, error) {
	var request = &protoscommon.QueryContract{}
	_, err := GetParam(request, in)
	if err != nil {
		return nil, rpcerror.WrongParam.GetStatus(), err
	}

	reply, err := r.workerQueryBytes(request.Driver, request.FuncName, request.Param)
	if err != nil {
		return nil, -1, err
	}
	if !reply.IsOk {
		return nil, -1, fmt.Errorf(string(reply.Msg))
	}
	return reply.Msg, rpcerror.Success.GetStatus(), nil
}

func (r *RPCMethodImpl) workerQueryMq(driver string, funcName string, param proto.Message) (*ptypes.Reply, error) {
	buf, err := util.Struct2Bytes(param)
	if err != nil {
		return nil, errors.New("marshal param failed")
	}
	return r.workerQueryBytes(driver, funcName, buf)
}

func (r *RPCMethodImpl) workerQueryBytes(driver string, funcName string, param []byte) (*ptypes.Reply, error) {
	data := &types.QueryExecutor{
		Driver:    driver,
		FuncName:  funcName,
		StateHash: nil,
		Param:     param,
	}
	resp, err := utils.SendMqRequestWithResp(r.module.workerClient, types.WorkerModule, mq.ModeWorker.String(), types.EventQueryContract, data, r.requestTimeOut)
	if err != nil {
		return nil, err
	}
	reply, ok := resp.Data.(*ptypes.Reply)
	if !ok {
		return nil, errors.New(rpcerror.IllegalResponse.GetMessage())
	}
	return reply, err
}

/*func completeResult(result interface{}, defaultResult interface{}) interface{} {
	if result == nil || reflect.TypeOf(result).Kind() != reflect.TypeOf(defaultResult).Kind() {
		return defaultResult
	}
	return result
}*/
